package com.shelpe.services.sinbad.service.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.shelpe.services.sinbad.controller.EffectableKeywordDto;
import com.shelpe.services.sinbad.controller.InvokesChunk;
import com.shelpe.services.sinbad.controller.KeywordsController.PricingKeyword;
import com.shelpe.services.sinbad.controller.dto.redis.RedisDto;
import com.shelpe.services.sinbad.controller.SearchCriteria;
import com.shelpe.services.sinbad.dao.WriteKeywordMapper;
import com.shelpe.services.sinbad.entity.jpa.Adgroup;
import com.shelpe.services.sinbad.entity.jpa.Campaign;
import com.shelpe.services.sinbad.entity.jpa.KeywordsLog;
import com.shelpe.services.sinbad.entity.jpa.OnlineStatus;
import com.shelpe.services.sinbad.entity.jpa.User;
import com.shelpe.services.sinbad.entity.jpa.vo.DisplayKeywordVo;
import com.shelpe.services.sinbad.entity.jpa.vo.KeywordsVo;
import com.shelpe.services.sinbad.entity.mongo.KeywordAddErrorLog;
import com.shelpe.services.sinbad.entity.mongo.KeywordDeleteErrorLog;
import com.shelpe.services.sinbad.entity.mongo.KeywordPriceErrorLog;
import com.shelpe.services.sinbad.model.Keyword;
import com.shelpe.services.sinbad.model.KeywordExample;
import com.shelpe.services.sinbad.model.KeywordID;
import com.shelpe.services.sinbad.model.RptKeyword;
import com.shelpe.services.sinbad.model.RptKeywordExample;
import com.shelpe.services.sinbad.readdao.KeywordMapper;
import com.shelpe.services.sinbad.readdao.ReadRptKeywordMapper;
import com.shelpe.services.sinbad.repository.jpa.AdgroupRepository;
import com.shelpe.services.sinbad.repository.jpa.CampaignRepository;
import com.shelpe.services.sinbad.repository.jpa.KeywordsLogRepository;
import com.shelpe.services.sinbad.repository.jpa.UserRepository;
import com.shelpe.services.sinbad.repository.mongo.KeywordAddErrorLogRepository;
import com.shelpe.services.sinbad.repository.mongo.KeywordDeleteErrorLogRepository;
import com.shelpe.services.sinbad.repository.mongo.KeywordPriceErrorLogRepository;
import com.shelpe.services.sinbad.service.SubscribleScheduler;
import com.shelpe.services.sinbad.service.SubscribleScheduler.Handler;
import com.shelpe.services.sinbad.service.SubscribleScheduler.SchedulerInterruptExcepion;
import com.shelpe.services.sinbad.service.mq.ShelpeMQProducerService;
import com.shelpe.services.sinbad.service.simbaproxy.AuthClient;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.AddKeywordsResponse;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.CreateKeywordDto;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.DelKeywordDto;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.DeleteKeywordsResponse;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.PriceKeywordsResponse;
import com.shelpe.services.sinbad.service.simbaproxy.SimbaProxyClient.UpdateKeywordDto;
import com.shelpe.services.sinbad.utils.CriteriaUtils;

import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import rx.Observable;
import rx.functions.Action1;

@Service
@Slf4j
@Setter
public class OopKeywordService {

	@Autowired
	private ObjectMapper objectMapper;
	@Autowired
	private WriteKeywordMapper keywordMapper;
	@Autowired
	private KeywordMapper readOnlyKeywordMapper;
	@Autowired
	private ReadRptKeywordMapper rptKeywordMapper;
	@Autowired
	private SimbaProxyClient simbaProxyClient;
	@Autowired
	private KeywordsLogRepository keywordsLogRepository;
	@Autowired
	private UserRepository userRepository;
	@Autowired
	private AdgroupRepository adgroupRepository;
	@Autowired
	private CampaignRepository campaignRepository;
	@Autowired
	private KeywordAddErrorLogRepository keywordAddErrorLogRepository;
	@Autowired
	private KeywordPriceErrorLogRepository keywordPriceErrorLogRepository;
	@Autowired
	private KeywordDeleteErrorLogRepository keywordDeleteErrorLogRepository;
	@Autowired
	private ShelpeMQProducerService mqProducerService;
	@Autowired
	private AuthClient authClient;

	public void price(final User user, List<PricingKeyword> keywords, Callback lastCallback, final boolean isManual,
			final boolean isSemiauto) {
		InvokesChunk<PricingKeyword> chunk = new InvokesChunk<PricingKeyword>();
		chunk.setLimit(50);
		List<Long> keywordIDs = new ArrayList<Long>();
		for (PricingKeyword keyword : keywords) {
			if (keyword.getKeywordID() > 0) {
				chunk.add(keyword);
				keywordIDs.add(keyword.getKeywordID());
			}
		}
		if (keywordIDs.size() == 0) {
			lastCallback.call(new ArrayList<Keyword>());
			return;
		}

		KeywordExample ex = new KeywordExample();
		ex.setUserId(user.getUserID());
		ex.createCriteria().andKeywordIdIn(keywordIDs);
		List<Keyword> originalKeywords = this.keywordMapper.selectByExample(ex);

		final AtomicInteger counter = new AtomicInteger();
		final List<Keyword> savedKeywords = new ArrayList<Keyword>();
		final KeywordsLogRepository logRepository = this.keywordsLogRepository;
		while (chunk != null) {
			this.pricingPerChunk(chunk, new InvokesChunk.ChunkCallback<PricingKeyword, Keyword>() {
				@Override
				public void call(List<PricingKeyword> inputs, List<Keyword> results) {
					if (results != null) {
						savedKeywords.addAll(results);
						// 保存操作记录
						if (results.size() > 0) {
							List<KeywordsLog> kwLogs = new ArrayList<KeywordsLog>();
							KeywordsLog.Action action = isManual ? KeywordsLog.Action.MANUAL_CHANGE_PRICE
									: KeywordsLog.Action.AUTO_CHANGE_PRICE;
							if (isSemiauto) {
								action = KeywordsLog.Action.SEMIAUTO_EXEC_UPDATE;
							}
							/* 判断是否是匹配方式的修改操作 */
							boolean isMatchScope = checkMatchScopeModel(results, originalKeywords);
							if (isMatchScope) {
								action = isManual ? KeywordsLog.Action.MANUAL_CHANGE_MATCHSCOPE
										: KeywordsLog.Action.AUTO_CHANGE_MATCHSCOPE;
							}

							Map<Long, Adgroup> cache = new HashMap<Long, Adgroup>();
							fillupDiscountMobilePrice(originalKeywords, cache);
							fillupDiscountMobilePrice(results, cache);

							for (Keyword changedKeyword : results) {
								Keyword originalKeyword = filterByKeywordID(originalKeywords,
										changedKeyword.getKeywordId());
								if (originalKeyword != null) {
									KeywordsLog kwLog = convert2Log(originalKeyword, changedKeyword, action,user.getUserID());
									kwLogs.add(kwLog);
								}
							}
							if (kwLogs.size() > 0) {
								logRepository.save(kwLogs);
//								try {
////									mqProducerService.sendKeywordsLog(objectMapper.writeValueAsString(kwLogs));
//								} catch (JsonProcessingException e) {
//									log.error("keywordsLogJsonProcessingException:{}", e);
//								}
							}
						}
					}

					// 最后一次调用
					if (counter.decrementAndGet() == 0) {
						log.debug("the last callback for priceKeywords", user);
						lastCallback.call(savedKeywords);
					}
				}
			}, user);
			chunk = chunk.getNext();
			counter.incrementAndGet();
		}
	}

	private boolean checkMatchScopeModel(List<Keyword> results, List<Keyword> originalKeywords) {
		Set<Keyword> set = new HashSet<Keyword>();
		Random random = new Random();
		/* 随机选取3个关键词进行比较,来判断价格是否改变,从而确认是否是匹配模式 */
		/* 遍历3次,有可能数据重复(重复不显示) */
		for (int i = 0; i < 3; i++) {
			set.add(results.get(random.nextInt(results.size())));
		}
		int trueNum = 0;
		for (Keyword kw : set) {
			for (Keyword oldKw : originalKeywords) {
				if (kw.getKeywordId().longValue() == oldKw.getKeywordId().longValue()
						&& kw.getMaxMobilePrice().intValue() == oldKw.getMaxMobilePrice().intValue()
						&& kw.getMaxPrice().intValue() == oldKw.getMaxPrice().intValue()) {
					if (kw.getMatchscope().intValue() != oldKw.getMatchscope().intValue()) {
						trueNum++;
					}
				}
			}
		}
		if (trueNum > 0) {
			return true;
		}
		return false;
	}

	private void fillupDiscountMobilePrice(List<Keyword> keywords, Map<Long, Adgroup> cache) {
		List<Long> adgroupIDs = new ArrayList<Long>();
		for (Keyword kw : keywords) {
			if (adgroupIDs.indexOf(kw.getAdgroupId()) == -1 && !cache.containsKey(kw.getAdgroupId())) {
				adgroupIDs.add(kw.getAdgroupId());
			}
		}
		if (adgroupIDs.size() > 0) {
			List<Adgroup> adgroups = adgroupRepository.findByAutoIDIn(adgroupIDs);
			if (adgroups != null) {
				for (Adgroup adgroup : adgroups) {
					cache.put(adgroup.getAutoID(), adgroup);
				}
			}
		}

		for (Keyword kw : keywords) {
			Adgroup adgroup = cache.get(kw.getAdgroupId());
			if (kw.getMobileIsDefaultPrice() == 1 && kw.getMaxPrice() > 0 && adgroup != null
					&& adgroup.getMobileDiscount() > 0) {
				kw.setMaxMobilePrice(kw.getMaxPrice() * adgroup.getMobileDiscount() / 100);
			}
		}
	}

	public void add(Adgroup adgroup, final List<PricingKeyword> keywords, final Callback lastCallback,
			final boolean isManual) {
		AuthClient.UserToken token = this.authClient.findSessionKeyByID(adgroup.getUserID());
		Assert.isTrue(token != null && token.isExpired(), "invalid-sessionkey");
		
		boolean check = false;
		for (PricingKeyword pkw : keywords) {
			if (pkw.getMaxMobilePrice() > 0) {
				check = true;
				break;
			}
		}
		final boolean needSetMobilePrice = check;
		final User user = this.userRepository.findOne(adgroup.getUserID());
		Assert.notNull(user, "user_not_found");

		this.addInternal(adgroup, keywords, new Callback() {
			@Override
			public void call(List<Keyword> savedKeywords) {
				// 如果需要设置移动价格，需要做一次调价操作
				if (needSetMobilePrice && savedKeywords != null && savedKeywords.size() > 0) {
					List<PricingKeyword> toPriceKeywords = new ArrayList<PricingKeyword>();
					for (PricingKeyword pkw : keywords) {
						Keyword addedKeyword = filterByWord(savedKeywords, pkw.getWord());
						if (addedKeyword != null) {
							pkw.setKeywordID(addedKeyword.getKeywordId());
							pkw.setMobileIsDefaultPrice(0);
							if (pkw.getMatchScope() == -1) {
								pkw.setMatchScope(pkw.getMatchScope());
							}
							toPriceKeywords.add(pkw);
						}
					}
					if (toPriceKeywords.size() > 0) {
						price(user, keywords, lastCallback, isManual, false);
					} else {
						log.warn("no mobile price to set: {}", adgroup);
					}
				}
				lastCallback.call(savedKeywords);
			}
		}, isManual);
	}

	private void addInternal(final Adgroup adgroup, List<PricingKeyword> keywords, Callback lastCallback,
			boolean isManual) {
		InvokesChunk<PricingKeyword> chunk = new InvokesChunk<PricingKeyword>();
		chunk.setLimit(80);
		chunk.setMaxLenth(Integer.MAX_VALUE);
		for (PricingKeyword keyword : keywords) {
			chunk.add(keyword);
		}

		final List<PricingKeyword> toPriceKeywords = keywords;
		final List<Keyword> savedKeywords = new ArrayList<Keyword>();
		final KeywordsLogRepository logRepository = this.keywordsLogRepository;
		final KeywordAddErrorLog errorLog = new KeywordAddErrorLog();// mongodb添加关键词错误表
		errorLog.from(adgroup);
		// 查询该宝贝下已经添加的词
		KeywordExample ex = new KeywordExample();
		ex.setUserId(adgroup.getUserID());
		ex.createCriteria().andAdgroupIdEqualTo(adgroup.getAutoID());
		List<Keyword> addedKeywords = this.keywordMapper.selectByExample(ex);
		if (addedKeywords != null && addedKeywords.size() > 0) {
			// 已经添加的关键词存入mongodb表中
			for (Keyword kw : addedKeywords) {
				KeywordAddErrorLog.RecommendWord rw = new KeywordAddErrorLog.RecommendWord();
				rw.setWord(kw.getWord());
				rw.setMatchScope(kw.getMatchscope());
				rw.setMaxPrice(kw.getMaxPrice());
				errorLog.addAddedWord(rw);
			}
		}

		SubscribleScheduler<SimbaProxyClient.AddKeywordsResponse> ssr = new SubscribleScheduler<SimbaProxyClient.AddKeywordsResponse>();
		ssr.setupConcurrency(1);
		ssr.setupHandler(new Handler<AddKeywordsResponse>() {
			@Override
			public void handle(AddKeywordsResponse res) {
				long start = System.currentTimeMillis();
				log.debug("addKeywords: {}", res);
				List<Keyword> keywords = new ArrayList<Keyword>();
				if (res != null && res.getRes() != null && res.getRes().getDataList() != null
						&& res.getRes().getDataList().getKeywords() != null) {
					for (SimbaProxyClient.Keyword kw : res.getRes().getDataList().getKeywords()) {
						Keyword keyword = kw.toPO(new Keyword());
						keyword.setMobileIsDefaultPrice(1);
						if (adgroup.getMobileDiscount() > 0) {
							keyword.setMaxMobilePrice(keyword.getMaxPrice() * adgroup.getMobileDiscount() / 100);
						}
						keyword.setUserId(adgroup.getUserID());
						keyword.setNick(adgroup.getNick());
						keyword.setUpdatedTime(new Date());
						keyword.setOrder((short) 0);
						keyword.setQscore(0);
						keyword.setSmartPrice(false);
						keyword.setWirelessQscore(0);
						keyword.setRankScore((short) 0);
						keywordMapper.insertSelective(keyword);
						keywords.add(keyword);
					}
				} else {
					if (res != null && res.getError() != null) {
						try {
							errorLog.setErrorCode(objectMapper.writeValueAsString(res.getError()));
						} catch (JsonProcessingException e) {
							errorLog.setErrorCode("unknown-error");
							log.error("JsonProcessingException:{}", e);
						}
						if (res.getError().getSubCode() != null) {
							errorLog.setSubCode(res.getError().getSubCode());
						}
						if (res.getError().getSubMessage() != null) {
							errorLog.setSubMessage(res.getError().getSubMessage());
						}
					} else {
						errorLog.setErrorCode("unknown-error");
					}

					for (PricingKeyword kw : toPriceKeywords) {
						KeywordAddErrorLog.RecommendWord rw = new KeywordAddErrorLog.RecommendWord();
						rw.setWord(kw.getWord());
						rw.setMatchScope(kw.getMatchScope());
						rw.setMaxPrice(kw.getMaxPrice());
						errorLog.addErrorWord(rw);
					}
					keywordAddErrorLogRepository.save(errorLog); // 保存记录到mongodb中
				}
				//log.info("add keywords:{}", keywords);
				if (keywords != null) {
					// 成功添加的关键词存入mongodb表中,保留下次测试
					/*
					 * for (Keyword kw : keywords) {
					 * KeywordAddErrorLog.RecommendWord rw = new
					 * KeywordAddErrorLog.RecommendWord();
					 * rw.setWord(kw.getWord());
					 * rw.setMatchScope(kw.getMatchscope());
					 * rw.setMaxPrice(kw.getMaxPrice());
					 * errorLog.addSuccessWord(rw); }
					 */
					final List<Keyword> results = keywords;
					savedKeywords.addAll(results);
					// 保存操作记录
					if (results.size() > 0) {
						List<KeywordsLog> kwLogs = new ArrayList<KeywordsLog>();
						KeywordsLog.Action action = isManual ? KeywordsLog.Action.MANUAL_NEW_KEYWORD
								: KeywordsLog.Action.AUTO_NEW_KEYWORD;
						Map<Long, Adgroup> cache = new HashMap<Long, Adgroup>();
						cache.put(adgroup.getAutoID(), adgroup);
						fillupDiscountMobilePrice(results, cache);
						for (Keyword kw : results) {
							KeywordsLog kwLog = convert2Log(null, kw, action,adgroup.getUserID());
							kwLogs.add(kwLog);
						}
						if (kwLogs.size() > 0) {
							logRepository.save(kwLogs);
//							try {
//								mqProducerService.sendKeywordsLog(objectMapper.writeValueAsString(kwLogs));
//							} catch (JsonProcessingException e) {
//								log.error("keywordsLogJsonProcessingException:{}", e);
//							}
						}
					}
				}
				log.info("addKeywords per call cost {}", System.currentTimeMillis() - start);
			}

		});
		while (chunk != null) {
			List<SimbaProxyClient.AddingKeyword> kws = new ArrayList<SimbaProxyClient.AddingKeyword>();
			for (PricingKeyword keyword : chunk.getWords()) {
				SimbaProxyClient.AddingKeyword kw = new SimbaProxyClient.AddingKeyword();
				BeanUtils.copyProperties(keyword, kw);
				kws.add(kw);
			}
			String kwString = this.serializeAddingKeyword(kws);
			log.debug("add words: {}", kwString);
			Assert.notNull(kwString, "fail to serialize the keywords");
			CreateKeywordDto dto = new CreateKeywordDto();
			dto.setAdgroupID(adgroup.getAutoID());
			dto.setNick(adgroup.getNick());
			dto.setKeywordPrices(kwString);
			Observable<SimbaProxyClient.AddKeywordsResponse> observer = this.simbaProxyClient.createKeywords(dto);

			ssr.addObservable(observer);
			chunk = chunk.getNext();
		}
		/* 最后一次调用 */
		ssr.invoke(new SubscribleScheduler.Callback() {
			@Override
			public void calll(SchedulerInterruptExcepion ex) {
				lastCallback.call(savedKeywords);
			}
		});
	}

	public void delete(final Campaign campaign, List<Long> keywordIDs, Callback lastCallback, final boolean isManual,
			final boolean isSemiauto) {
		InvokesChunk<DeletingKeyword> chunk = new InvokesChunk<DeletingKeyword>();
		chunk.setLimit(100);
		chunk.setMaxLenth(Integer.MAX_VALUE);
		for (long id : keywordIDs) {
			DeletingKeyword keyword = new DeletingKeyword();
			keyword.setKeywordID(id);
			chunk.add(keyword);
		}
		if (keywordIDs.size() == 0) {
			lastCallback.call(new ArrayList<Keyword>());
			return;
		}

		KeywordExample ex = new KeywordExample();
		ex.setUserId(campaign.getUserID());
		ex.createCriteria().andKeywordIdIn(keywordIDs);
		List<Keyword> originalKeywords = this.keywordMapper.selectByExample(ex);

		final AtomicInteger counter = new AtomicInteger();
		final List<Keyword> savedKeywords = new ArrayList<Keyword>();
		final KeywordsLogRepository logRepository = this.keywordsLogRepository;
		while (chunk != null) {
			this.deletePerChunk(chunk, new InvokesChunk.ChunkCallback<DeletingKeyword, Keyword>() {
				@Override
				public void call(List<DeletingKeyword> inputs, List<Keyword> results) {
					if (results != null) {
						savedKeywords.addAll(results);

						// 保存操作记录
						if (results.size() > 0) {
							List<KeywordsLog> kwLogs = new ArrayList<KeywordsLog>();
							KeywordsLog.Action action = isManual ? KeywordsLog.Action.MANUAL_DEL_KEYWORD
									: KeywordsLog.Action.AUTO_DEL_KEYWORD;
							if (isSemiauto) {
								action = KeywordsLog.Action.SEMIAUTO_EXEC_DEL;
							}
							Map<Long, Adgroup> cache = new HashMap<Long, Adgroup>();
							fillupDiscountMobilePrice(originalKeywords, cache);
							for (Keyword deletedKeyword : results) {
								Keyword kw = filterByKeywordID(originalKeywords, deletedKeyword.getKeywordId());
								if (kw != null) {
									KeywordsLog kwLog = convert2Log(kw, null, action,campaign.getUserID());
									kwLogs.add(kwLog);
								}
							}
							if (kwLogs.size() > 0) {
								logRepository.save(kwLogs);
//								try {
//									mqProducerService.sendKeywordsLog(objectMapper.writeValueAsString(kwLogs));
//								} catch (JsonProcessingException e) {
//									log.error("keywordsLogJsonProcessingException:{}", e);
//								}
							}
						}
					}

					// 最后一次调用
					if (counter.decrementAndGet() == 0) {
						log.debug("the last callback for deleteKeywords", campaign);
						lastCallback.call(savedKeywords);
					}
				}
			}, campaign);
			chunk = chunk.getNext();
			counter.incrementAndGet();
		}
	}

	public static interface Callback {
		void call(List<Keyword> keywords);
	}

	private Keyword filterByKeywordID(List<Keyword> keywords, long keywordID) {
		for (Keyword kw : keywords) {
			if (kw.getKeywordId() == keywordID) {
				return kw;
			}
		}
		return null;
	}

	private Keyword filterByWord(List<Keyword> keywords, String word) {
		for (Keyword kw : keywords) {
			if (word.equals(kw.getWord())) {
				return kw;
			}
		}
		return null;
	}

	private void deletePerChunk(final InvokesChunk<DeletingKeyword> chunk,
			final InvokesChunk.ChunkCallback<DeletingKeyword, Keyword> next, final Campaign campaign) {
		if (chunk.getWords() == null || chunk.getWords().size() == 0) {
			next.call(chunk.getWords(), null);
			return;
		}

		List<Long> keywordIDs = new ArrayList<Long>();
		for (DeletingKeyword keyword : chunk.getWords()) {
			keywordIDs.add(keyword.getKeywordID());
		}

		DelKeywordDto dto = new DelKeywordDto();
		dto.setCampaignID(campaign.getAutoID());
		dto.setNick(campaign.getNick());
		dto.setKeywordIDs(new ArrayList<Long>());
		dto.getKeywordIDs().addAll(keywordIDs);

		Observable<SimbaProxyClient.DeleteKeywordsResponse> observer = this.simbaProxyClient.removeKeywords(dto);
		log.debug("delete set post:{}", dto);
		observer.subscribe(new Action1<SimbaProxyClient.DeleteKeywordsResponse>() {
			@Override
			public void call(DeleteKeywordsResponse res) {
				log.debug("deleteKeywords: {}", res);
				long start = System.currentTimeMillis();
				List<Keyword> keywords = new ArrayList<Keyword>();
				if (res != null && res.getRes() != null && res.getRes().getDataList() != null
						&& res.getRes().getDataList().getKeywords() != null) {
					List<Long> deletedKeywordIDs = new ArrayList<Long>();
					for (SimbaProxyClient.Keyword kw : res.getRes().getDataList().getKeywords()) {
						deletedKeywordIDs.add(kw.getKeywordId());
						Keyword keyword = kw.toPO(new Keyword());
						keyword.setUserId(campaign.getUserID());
						keyword.setNick(campaign.getNick());
						keyword.setUpdatedTime(new Date());
						keywords.add(keyword);
					}
					if (deletedKeywordIDs.size() > 0) {
						KeywordExample ex = new KeywordExample();
						ex.setUserId(campaign.getUserID());
						ex.createCriteria().andKeywordIdIn(keywordIDs);
						keywordMapper.deleteByExample(ex);
					}
				} else {
					log.error("fail to sync taobao when deleteKeywords: {} - {}", res != null ? res.getError() : "",
							keywordIDs);
					KeywordDeleteErrorLog errorLog = new KeywordDeleteErrorLog();
					errorLog.from(campaign);
					if (res != null) {
						try {
							errorLog.setRes(objectMapper.writeValueAsString(res));
						} catch (JsonProcessingException e1) {
							log.error("JsonProcessingException:{}", e1);
							errorLog.setErrorCode("unknown-error");
						}
						if (res.getError() != null) {
							try {
								errorLog.setErrorCode(objectMapper.writeValueAsString(res.getError()));
							} catch (JsonProcessingException e) {
								errorLog.setErrorCode("unknown-error");
								log.error("JsonProcessingException:{}", e);
							}
							if (res.getError().getSubCode() != null) {
								errorLog.setSubCode(res.getError().getSubCode());
							}
							if (res.getError().getSubMessage() != null) {
								errorLog.setSubMessage(res.getError().getSubMessage());
							}
						} else {
							errorLog.setErrorCode("unknown-error");
						}
					}
					for (Long keywordID : keywordIDs) {
						errorLog.addKeyWordID(keywordID);
					}
					keywordDeleteErrorLogRepository.save(errorLog);
				}
				next.call(chunk.getWords(), keywords);
				log.info("deleteKeywords per call cost {}", System.currentTimeMillis() - start);
			}
		}, new Action1<Throwable>() {
			@Override
			public void call(Throwable t) {
				log.error("catch exception when deleteKeywords: {} - {}", keywordIDs, t);
				next.call(chunk.getWords(), null);
			}
		});
	}

	private void pricingPerChunk(final InvokesChunk<PricingKeyword> chunk,
			final InvokesChunk.ChunkCallback<PricingKeyword, Keyword> next, final User user) {
		if (chunk.getWords() == null || chunk.getWords().size() == 0) {
			next.call(chunk.getWords(), null);
			return;
		}

		List<SimbaProxyClient.HandlingKeyword> kws = new ArrayList<SimbaProxyClient.HandlingKeyword>();
		for (PricingKeyword keyword : chunk.getWords()) {
			SimbaProxyClient.HandlingKeyword kw = new SimbaProxyClient.HandlingKeyword();
			BeanUtils.copyProperties(keyword, kw);
			kws.add(kw);
		}
		String kwString = this.serializeHandlingKeyword(kws);
		log.debug("pricing words: {}", kwString);
		Assert.notNull(kwString, "fail to serialize the keywords");

		UpdateKeywordDto dto = new UpdateKeywordDto();
		dto.setNick(user.getNick());
		dto.setJsonString(kwString);
		Observable<SimbaProxyClient.PriceKeywordsResponse> observer = this.simbaProxyClient
				.updateKeywords(dto);
		observer.subscribe(new Action1<SimbaProxyClient.PriceKeywordsResponse>() {
			@Override
			public void call(PriceKeywordsResponse res) {
				log.debug("priceKeywords: {}", res);
				long start = System.currentTimeMillis();
				List<Keyword> keywords = new ArrayList<Keyword>();
				if (res != null && res.getRes() != null && res.getRes().getDataList() != null
						&& res.getRes().getDataList().getKeywords() != null) {
					for (SimbaProxyClient.Keyword kw : res.getRes().getDataList().getKeywords()) {
						Keyword keyword = kw.toPO(new Keyword());
						keyword.setMatchscope(kw.getMatchscope());
						keyword.setMaxPrice(kw.getMaxPrice());
						keyword.setIsDefaultPrice(kw.isDefaultPrice());
						keyword.setMaxMobilePrice(kw.getMaxMobilePrice());
						keyword.setMobileIsDefaultPrice(kw.getMobileIsDefaultPrice());
						keyword.setUserId(user.getUserID());
						keyword.setKeywordId(kw.getKeywordId());
						keyword.setQscore(null);
						keywordMapper.updateByPrimaryKeySelective(keyword);
						keywords.add(keyword);
					}
				} else {
					log.error("fail to sync taobao when priceKeywords: {} - {}", res != null ? res.getError() : "",
							kws);
					KeywordPriceErrorLog errorLog = new KeywordPriceErrorLog();
					errorLog.from(user);
					if (res != null) {
						try {
							errorLog.setRes(objectMapper.writeValueAsString(res));
						} catch (JsonProcessingException e) {
							log.error("JsonProcessingException:{}", e);
							errorLog.setRes("unknown-error");
						}
					}
					errorLog.setErrorCode(
							(res != null && res.getError() != null) ? res.getError().getSubMessage() : "unknown-error");
					for (SimbaProxyClient.HandlingKeyword hk : kws) {
						KeywordPriceErrorLog.KeywordWord kw = new KeywordPriceErrorLog.KeywordWord();
						BeanUtils.copyProperties(hk, kw);
						errorLog.addWord(kw);
					}
					try {
						keywordPriceErrorLogRepository.save(errorLog);
					} catch (DuplicateKeyException ex) {
						log.error("DuplicateKeyException: {}", ex);
					}
				}
				next.call(chunk.getWords(), keywords);
				log.info("priceKeywords per call cost {}", System.currentTimeMillis() - start);
			}
		}, new Action1<Throwable>() {
			@Override
			public void call(Throwable t) {
				log.error("catch exception when priceKeywords: {} - {}", kws, t);
				next.call(chunk.getWords(), null);
			}
		});
	}

	// private void addingPerChunk(final InvokesChunk<PricingKeyword> chunk,
	// final SubscribleScheduler<SimbaProxyClient.AddKeywordsResponse> ssr,
	// final InvokesChunk.ChunkCallback<PricingKeyword, Keyword> callback, final
	// Adgroup adgroup) {
	// if (chunk.getWords() == null || chunk.getWords().size() == 0) {
	// callback.call(chunk.getWords(), null);
	// return;
	// }
	//
	// }

	private String serializeAddingKeyword(List<SimbaProxyClient.AddingKeyword> keywords) {
		try {
			return this.objectMapper.writer().writeValueAsString(keywords);
		} catch (JsonProcessingException e) {
			log.error("JsonProcessingException", e);
		}
		return null;
	}

	private String serializeHandlingKeyword(List<SimbaProxyClient.HandlingKeyword> keywords) {
		try {
			return this.objectMapper.writer().writeValueAsString(keywords);
		} catch (JsonProcessingException e) {
			log.error("JsonProcessingException", e);
		}
		return null;
	}

	private KeywordsLog convert2Log(Keyword original, Keyword latest, KeywordsLog.Action action,long userID) {
		KeywordsLog kwLog = new KeywordsLog();
		kwLog.setUpdatedTime(new Date());
		kwLog.setAction(action);

		DisplayKeywordVo note = new DisplayKeywordVo();
		if (latest != null) {
			kwLog.setAdgroupID(latest.getAdgroupId());
			kwLog.setKeywordID(latest.getKeywordId());
			kwLog.setKeyword(latest.getWord());
			kwLog.setUserID(userID);
			kwLog.setCampaignID(latest.getCampaignId());
			KeywordsVo.Vo newNote = new KeywordsVo.Vo();
			BeanUtils.copyProperties(latest, newNote);
			note.setNewNote(newNote);
		}
		if (original != null) {
			kwLog.setAdgroupID(original.getAdgroupId());
			kwLog.setKeywordID(original.getKeywordId());
			kwLog.setKeyword(original.getWord());
			kwLog.setUserID(userID);
			kwLog.setCampaignID(original.getCampaignId());
			KeywordsVo.Vo oldNote = new KeywordsVo.Vo();
			BeanUtils.copyProperties(original, oldNote);
			note.setOldNote(oldNote);
		}

		String jsonStr = "";
		try {
			jsonStr = objectMapper.writer().writeValueAsString(note);
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
		kwLog.setNote(jsonStr);
		return kwLog;
	}

	@Data
	public static class DeletingKeyword implements InvokesChunk.Lengthable {
		private long keywordID;

		@Override
		public int length() {
			return 50;
		}
	}

	@Data
	public static class SearchSettings {
		private boolean onlineAdgroup;
		private List<SearchCriteria> criterias;
	}

	public List<EffectableKeywordDto> findKeywords(long userID, List<Long> adgroupIDs, List<Long> campaignIDs,
			String device, int days, SearchSettings settings) {
		List<EffectableKeywordDto> rets = new ArrayList<EffectableKeywordDto>();
		long start = System.currentTimeMillis();

		// 分解成推广组来查询
		List<Adgroup> adgroups = new ArrayList<Adgroup>();
		if (adgroupIDs != null && adgroupIDs.size() > 0) {
			adgroups.addAll(this.adgroupRepository.findByAutoIDIn(adgroupIDs));
		}
		if (campaignIDs != null && campaignIDs.size() > 0) {
			if (settings.onlineAdgroup) {// 如果需要推广中的推广组，可以通过推广中推广计划来过滤一些无用的
				List<Campaign> campaigns = this.campaignRepository.findByAutoIDIn(campaignIDs);
				for (Campaign cam : campaigns) {
					if (cam.getOnlineStatus() != OnlineStatus.online) {
						campaignIDs.remove(cam.getAutoID());
					}
				}
			}
			if (campaignIDs.size() > 0) {
				adgroups.addAll(this.adgroupRepository.findByCampaignIDIn(campaignIDs));
			}
		}

		// 得到符合的推广组编号
		List<Long> selectedAdgroupIDs = new ArrayList<Long>();
		for (Adgroup ad : adgroups) {
			boolean canAdd = true;
			if (settings.onlineAdgroup) {
				canAdd = (ad.getOnlineStatus() == OnlineStatus.online) && ad.getOfflineType() == OnlineStatus.online;
			}
			if (canAdd) {
				selectedAdgroupIDs.add(ad.getAutoID());
			}
		}
		log.info("findKeywords - query keyword with adgroupIDs: {}", selectedAdgroupIDs);
		if (selectedAdgroupIDs.size() == 0) {
			return rets;
		}

		List<Long> keywordIDs = new ArrayList<Long>();
		List<KeywordID> kIDs = new ArrayList<KeywordID>();
		// 如果设置了查询条件，需要根据报表数据解决查询
		if (settings.getCriterias() != null && settings.getCriterias().size() > 0) {
			// 元数据查询
			KeywordExample example = new KeywordExample();
			example.setUserId(userID);
			KeywordExample.Criteria criteria = example.createCriteria();
			criteria.andAdgroupIdIn(selectedAdgroupIDs);

			boolean needSelectRpt = true;
			if (settings.getCriterias() != null) {
				for (SearchCriteria sc : settings.getCriterias()) {
					if (sc.getKey().equals("impressions") && (sc.getOperation() == SearchCriteria.Operation.lt.name()
							|| sc.getOperation() == SearchCriteria.Operation.lte.name()
							|| (sc.getOperation() == SearchCriteria.Operation.eq.name()
									&& sc.getValue().equals("0.0")))) {
						needSelectRpt = false;
					}
					CriteriaUtils.toCriteria(criteria, sc);
				}
			}
			List<KeywordID> kwIDs = readOnlyKeywordMapper.selectIDsByExample(example);
			log.info("findKeywords - find {} from keywords table", kwIDs.size());
			if (kwIDs.size() == 0) {
				return rets;
			} else {
				kIDs = kwIDs;
			}

			if (needSelectRpt) {
				// 报表查询
				RptKeywordExample rptExample = new RptKeywordExample();
				rptExample.setUserId(userID);
				RptKeywordExample.Criteria rptCriteria = this.buildRptKeywordExampleCriteria(rptExample, days, device);
				rptCriteria.andAdgroupIdIn(selectedAdgroupIDs);
				if (kwIDs.size() <= 100) {// 如果keywordID比较少，可以根据关键词查询
					List<Long> toQueryIDs = new ArrayList<Long>(150);
					for (KeywordID id : kwIDs) {
						toQueryIDs.add(id.getKeywordId());
					}
					rptCriteria.andKeywordIdIn(toQueryIDs);
				}
				if (settings.getCriterias() != null) {
					RptKeywordExample.Criteria rptHavingCriteria = rptExample.createHavingCriteria();
					for (SearchCriteria sc : settings.getCriterias()) {
						CriteriaUtils.toCriteria(rptHavingCriteria, sc);
					}
				}
				List<KeywordID> rptIDs = this.rptKeywordMapper.selectIDsByExampleGroupByKeywordID(rptExample);
				log.info("findKeywords - find {} from rptkeywords table", rptIDs.size());

				for (KeywordID id : rptIDs) {
					if (kwIDs.indexOf(id) > -1) {
						keywordIDs.add(id.getKeywordId());
					}
				}
			}
		} else {
			for (KeywordID keywordID : kIDs) {
				keywordIDs.add(keywordID.getKeywordId());
			}
		}

		// 查找完整的数据
		KeywordExample example = new KeywordExample();
		example.setUserId(userID);
		KeywordExample.Criteria criteria = example.createCriteria();
		if (keywordIDs.size() > 0) {
			criteria.andKeywordIdIn(keywordIDs);
		} else {
			criteria.andAdgroupIdIn(selectedAdgroupIDs);
		}
		List<Keyword> keywords = this.readOnlyKeywordMapper.selectByExample(example);
		RptKeywordExample rptExample = new RptKeywordExample();
		rptExample.setUserId(userID);
		RptKeywordExample.Criteria rptCriteria = this.buildRptKeywordExampleCriteria(rptExample, days, device);
		if (keywordIDs.size() > 0) {
			rptCriteria.andKeywordIdIn(keywordIDs);
		} else {
			rptCriteria.andAdgroupIdIn(selectedAdgroupIDs);
		}
		List<RptKeyword> rptKeywords = this.rptKeywordMapper.selectByExampleGroupByKeywordID(rptExample);

		// 根据推广组过滤
		for (Keyword kw : keywords) {
			EffectableKeywordDto dto = new EffectableKeywordDto();
			dto.from(kw);
			RptKeyword rpt = this.filterRptKeywordByKeywordID(rptKeywords, kw.getKeywordId());
			if (rpt != null) {
				dto.from(rpt);
			}

			boolean canAdd = false;
			Adgroup adgroup = this.filteAdgrouprByAdgorupID(adgroups, kw.getAdgroupId());
			if (adgroup != null) {
				// 移动溢价的计算
				if (dto.isAsDefaultMobilePrice()) {
					dto.setMaxMobilePrice(dto.getMaxPrice() * adgroup.getMobileDiscount() / 100);
				}
				if (settings.onlineAdgroup) {
					canAdd = (adgroup.getOnlineStatus() == OnlineStatus.online);
				} else {
					canAdd = true;
				}
			}
			if (canAdd) {
				rets.add(dto);
			}
		}

		log.info("findKeywords cost {} ms: userID({}) - campaignIDs({}) - adgroupIDs({})",
				System.currentTimeMillis() - start, userID, campaignIDs, adgroupIDs);
		return rets;
	}

	private RptKeywordExample.Criteria buildRptKeywordExampleCriteria(RptKeywordExample example, int days,
			String device) {
		RptKeywordExample.Criteria criteria = example.createCriteria();
		Date yesterday = DateUtils.addDays(DateUtils.truncate(new Date(), Calendar.DATE), -1);
		criteria.andDateBetween(DateUtils.addDays(yesterday, 1 - days), yesterday);

		if ("mobile".equals(device)) {
			criteria.andSourceBetween((byte) 4, (byte) 5);
		} else {
			criteria.andSourceBetween((byte) 1, (byte) 2);
		}
		return criteria;
	}

	private RptKeyword filterRptKeywordByKeywordID(List<RptKeyword> rpts, long keywordID) {
		for (RptKeyword rpt : rpts) {
			if (rpt.getKeywordId() == keywordID) {
				return rpt;
			}
		}
		return null;
	}

	private Adgroup filteAdgrouprByAdgorupID(List<Adgroup> adgroups, long adgroupID) {
		for (Adgroup adgroup : adgroups) {
			if (adgroup.getAutoID() == adgroupID) {
				return adgroup;
			}
		}
		return null;
	}
}
